package small.common.executeEngine.engine.delay;

import small.common.core.exceptions.CommonException;
import small.common.core.runnable.ExecuteRunnable;
import small.common.core.thread.executor.NameThreadFactory;
import small.common.executeEngine.engine.AbstractTaskExecuteEngine;
import small.common.executeEngine.processor.TaskProcessor;
import small.common.executeEngine.task.AbstractDelayTask;
import small.common.log.LogProvider;
import org.slf4j.Logger;

import java.util.Collection;
import java.util.HashSet;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.ReentrantLock;

public class DelayTaskExecuteEngine extends AbstractTaskExecuteEngine<AbstractDelayTask> {

    protected final ConcurrentHashMap<Object, AbstractDelayTask> tasks;

    protected final ScheduledExecutorService processingExecutor;

    protected final ReentrantLock lock = new ReentrantLock();

    private int maxRetryTimes = 0;

    public DelayTaskExecuteEngine(String name) {
        this(name, 32, LogProvider.getLogger(DelayTaskExecuteEngine.class), 100L, 3);
    }

    public DelayTaskExecuteEngine(String name, Logger logger) {
        this(name, 32, logger, 100L, 3);
    }

    public DelayTaskExecuteEngine(String name, int initCapacity, Logger logger) {
        this(name, initCapacity, logger, 100L, 3);
    }

    public DelayTaskExecuteEngine(String name, int initCapacity, Logger logger, long processInterval) {
        this(name, initCapacity, logger, processInterval, 3);
    }

    public DelayTaskExecuteEngine(String name, int initCapacity, Logger logger, long processInterval, int maxRetryTimes) {
        super(logger);
        this.maxRetryTimes = maxRetryTimes;
        tasks = new ConcurrentHashMap<Object, AbstractDelayTask>(initCapacity);
        this.processingExecutor = Executors.newScheduledThreadPool(1, new NameThreadFactory(name));
        //延时定时任务
        processingExecutor
                .scheduleWithFixedDelay(new DelayProcessRunnable(), processInterval, processInterval, TimeUnit.MILLISECONDS);
    }

    @Override
    public void shutdown() throws CommonException {
        processingExecutor.shutdown();
    }

    @Override
    public void addTask(Object key, AbstractDelayTask newTask) {
        AbstractDelayTask existTask = tasks.get(key);
        lock.lock();
        try {

            if (null != existTask) {
                newTask.merge(existTask);
            }
            tasks.put(key, newTask);
        } finally {
            lock.unlock();
        }
    }

    @Override
    public AbstractDelayTask removeTask(Object key) {
        AbstractDelayTask existTask = tasks.get(key);
        lock.lock();
        try {

            if (null != existTask) {
                return tasks.remove(key);
            }
            return null;
        } finally {
            lock.unlock();
        }
    }


    @Override
    public Collection<Object> getAllTaskKeys() {
        Collection<Object> keys = new HashSet<Object>();
        lock.lock();
        try {
            keys.addAll(tasks.keySet());
        } finally {
            lock.unlock();
        }
        return keys;
    }

    protected void processTasks() {
        Collection<Object> keys = getAllTaskKeys();
        for (Object taskKey : keys) {

            AbstractDelayTask task = removeTask(taskKey);
            if (null == task || !task.shouldProcess()) {
                continue;
            }
            TaskProcessor processor = getProcessor(taskKey);
            if (processor == null) {
                continue;
            }
            try {
                // 执行当前任务
                if (!processor.process(task)) {
                    //执行失败再次执行
                    retryFailedTask(taskKey, task);
                }
            } catch (Throwable e) {
                log.error("task execute error : " + e.toString(), e);
                //执行失败再次执行
                retryFailedTask(taskKey, task);
            }
        }
    }


    private void retryFailedTask(Object key, AbstractDelayTask task) {
        if (task.getRetryTimes() >= this.maxRetryTimes) {
            return;
        }
        task.increment();
        task.setLastProcessTime(System.currentTimeMillis());
        addTask(key, task);
    }

    private class DelayProcessRunnable extends ExecuteRunnable {
        @Override
        protected void executeBody() {
            processTasks();
        }
    }
}
